package com.alibaba.datax.plugin.writer.miniowriter.handler;

import com.alibaba.datax.common.element.Column;
import com.alibaba.datax.common.element.DateColumn;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.writer.miniowriter.Key;
import org.apache.commons.lang3.StringUtils;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author JDragon
 * @Date 2021.07.12 上午 11:35
 * @Email 1061917196@qq.com
 * @Des:
 */
public abstract class WriterHandler {

    protected Configuration configuration;

    protected TaskPluginCollector taskPluginCollector;

    protected RecordReceiver lineReceiver;

    private String nullFormat;

    private DateFormat dateParse;

    public void startWrite() {
        List<String> heads = configuration.getList(Key.HEADER, String.class);
        nullFormat = configuration.getString(Key.NULL_FORMAT, "");
        String dateFormat = configuration.getString(Key.DATE_FORMAT);
        if (StringUtils.isNotBlank(dateFormat)) {
            dateParse = new SimpleDateFormat(dateFormat);
        }

        writerHeader(heads);
        Record record;
        while ((record = lineReceiver.getFromReader()) != null) {
            write(record);
        }
        finish();
    }

    public List<String> resolveRow(Record record) {
        List<String> rowData = new ArrayList<>();
        int recordLength = record.getColumnNumber();
        for (int i = 0; i < recordLength; i++) {
            Column column = record.getColumn(i);
            if (column.getRawData() == null) {
                rowData.add(nullFormat);
            } else if (column instanceof DateColumn && dateParse != null) {
                rowData.add(dateParse.format(column.asDate()));
            } else {
                rowData.add(column.asString());
            }
        }
        return rowData;
    }

    public abstract void writerHeader(List<String> header);

    public abstract void write(Record record);

    public abstract void finish();

    public void setTaskPluginCollector(TaskPluginCollector taskPluginCollector) {
        this.taskPluginCollector = taskPluginCollector;
    }

    public void setLineReceiver(RecordReceiver lineReceiver) {
        this.lineReceiver = lineReceiver;
    }

    public void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
    }

    public void collectDirtyRecord(Record record, Throwable e) {
        taskPluginCollector.collectDirtyRecord(record, e);
    }
}
